package tools

/*
 +----------------------------------------------------------------------
 + Title        : ReliabilityProducer
 + Author       : huwl
 + Version      : 1.0.0
 + Initial-Time : 2021-09-22
 + Last-time    : 这个文件最后修改的时间 + 修改人的名称
 + Desc         : 生产者可靠性
 +----------------------------------------------------------------------
*/
import (
	"context"
	"fmt"
	"os"
	"os/signal"
	"syscall"
	"time"

	amqp "github.com/rabbitmq/amqp091-go"
)

/**
 * 生产者可靠性案列
 *
 * @param string url --------------------------------------------------必传
 * @param string vhost ------------------------------------------------必传
 * @param []byte msg --------------------------------------------------必传
 * @param string msgType ----------------------------------------------必传
 * @param <-chan struct{} publishOkCh ---------------------------------必传
 * @param chan<- *amqp.DeferredConfirmation confirmsCh ----------------必传
 * @param <-chan struct{} confirmsDoneCh ------------------------------必传
 * @param chan struct{} exitCh ----------------------------------------必传
 * @author huwl
 */
func ReliabilityProducer(url, vhost string, msg []byte, msgType string, publishOkCh <-chan struct{}, confirmsCh chan<- *amqp.DeferredConfirmation, confirmsDoneCh <-chan struct{}, exitCh chan struct{}) {
	continuous := false
	config := amqp.Config{
		Vhost:      vhost,
		Properties: amqp.NewConnectionProperties(),
	}

	config.Properties.SetClientConnectionName("producer-with-confirms")

	conn, err := amqp.DialConfig(url, config)
	if err != nil {
		fmt.Println("amqp.Dial err =", err.Error())
		return
	}

	defer conn.Close()

	fmt.Println("连接 rabbitMQ 成功")

	ch, err := conn.Channel()
	if err != nil {
		fmt.Println("conn.Channel err =", err.Error())
		return
	}

	defer ch.Close()

	// 设置管道确认模式
	if err := ch.Confirm(false); err != nil {
		fmt.Println("生产者管道无法进入确认模式错误：", err.Error())
		return
	}

	// 创建队列
	q, err := ch.QueueDeclare(
		"reliable.queue", // Queuename
		// 设置队列持久化
		true,  // durable：持久的
		false, // autoDelete：自动删除
		false, // exclusive：专用的
		false, // noWait：不等待、同步
		nil,   // arguments
	)
	if err != nil {
		fmt.Println("创建队列错误：", err.Error())
		return
	}

	for {
		canPublish := false
		for {
			select {
			case <-confirmsDoneCh:
				fmt.Println("producer: stopping, all confirms seen")
				return
			case <-publishOkCh:
				fmt.Println("producer: got the OK to publish")
				canPublish = true
			case <-time.After(time.Second):
				fmt.Println("producer: still waiting on the OK to publish...")
				continue
			}

			if canPublish {
				break
			}
		}

		// 设置超时时间
		ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
		defer cancel()

		dConfirmation, err := ch.PublishWithDeferredConfirmWithContext(ctx,
			"",     // exchange：交换机
			q.Name, // routing key：路由名
			false,  // mandatory：强制性的
			false,  // immediate：立刻的
			// 消息包
			amqp.Publishing{
				ContentType: msgType,
				Body:        msg,
				// 设置消息持久化
				DeliveryMode: amqp.Persistent,
				AppId:        "sequential-producer",
			})

		if err != nil {
			fmt.Println("发送消息错误: ", err)
			fmt.Println("")
		}

		select {
		case <-confirmsDoneCh:
			fmt.Println("producer: stopping, all confirms seen")
			return
		case confirmsCh <- dConfirmation:
			fmt.Println("producer: delivered deferred confirm to handler")
			break
		}

		select {
		case <-confirmsDoneCh:
			fmt.Println("producer: stopping, all confirms seen")
			return
		case <-time.After(time.Millisecond * 250):
			if continuous {
				continue
			} else {
				fmt.Println("producer: initiating stop")
				close(exitCh)
				select {
				case <-confirmsDoneCh:
					fmt.Println("producer: stopping, all confirms seen")
					return
				case <-time.After(time.Second * 10):
					fmt.Println("producer: may be stopping with outstanding confirmations")
					return
				}
			}
		}
	}
}

/**
 * 处理关闭
 *
 * @param chan struct{} exitCh -------------------------------必传
 * @author huwl
 */
func SetupCloseHandler(exitCh chan struct{}) {
	c := make(chan os.Signal, 2)
	signal.Notify(c, os.Interrupt, syscall.SIGTERM)
	go func() {
		<-c
		fmt.Println("close handler: Ctrl+C pressed in Terminal")
		close(exitCh)
	}()
}

/**
 * 开始处理确认
 *
 * @param chan<- struct{} publishOkCh ----------------------必传
 * @param <-chan *amqp.DeferredConfirmation confirmsCh -----必传
 * @param chan struct{} confirmsDoneCh ---------------------必传
 * @param <-chan struct{} exitCh -----------------------------必传
 * @author huwl
 */
func StartConfirmHandler(publishOkCh chan<- struct{}, confirmsCh <-chan *amqp.DeferredConfirmation, confirmsDoneCh chan struct{}, exitCh <-chan struct{}) {
	go func() {
		confirms := make(map[uint64]*amqp.DeferredConfirmation)

		for {
			select {
			case <-exitCh:
				ExitConfirmHandler(confirms, confirmsDoneCh)
				return
			default:
				break
			}

			outstandingConfirmationCount := len(confirms)

			// Note: 8 is arbitrary, you may wish to allow more outstanding confirms before blocking publish
			if outstandingConfirmationCount <= 8 {
				select {
				case publishOkCh <- struct{}{}:
					fmt.Println("confirm handler: sent OK to publish")
				case <-time.After(time.Second * 5):
					fmt.Println("confirm handler: timeout indicating OK to publish (this should never happen!)")
				}
			} else {
				fmt.Println("confirm handler: waiting on outstanding confirmations, blocking publish", outstandingConfirmationCount)
			}

			select {
			case confirmation := <-confirmsCh:
				dtag := confirmation.DeliveryTag
				confirms[dtag] = confirmation
			case <-exitCh:
				ExitConfirmHandler(confirms, confirmsDoneCh)
				return
			}

			CheckConfirmations(confirms)
		}
	}()
}

/**
 * 处理退出确认
 *
 * @param map[uint64]*amqp.DeferredConfirmation confirms ---必传
 * @param chan struct{} confirmsDoneCh ---------------------必传
 * @author huwl
 */
func ExitConfirmHandler(confirms map[uint64]*amqp.DeferredConfirmation, confirmsDoneCh chan struct{}) {
	fmt.Println("confirm handler: exit requested")
	WaitConfirmations(confirms)
	close(confirmsDoneCh)
	fmt.Println("confirm handler: exiting")
}

/**
 * 检查确认
 *
 * @param map[uint64]*amqp.DeferredConfirmation confirms -----必传
 * @author huwl
 */
func CheckConfirmations(confirms map[uint64]*amqp.DeferredConfirmation) {
	fmt.Println("confirm handler: checking outstanding confirmations", len(confirms))
	for k, v := range confirms {
		if v.Acked() {
			fmt.Println("confirm handler: confirmed delivery with tag:", k)
			delete(confirms, k)
		}
	}
}

/**
 * 等待确认
 *
 * @param map[uint64]*amqp.DeferredConfirmation confirms -----必传
 * @author huwl
 */
func WaitConfirmations(confirms map[uint64]*amqp.DeferredConfirmation) {
	fmt.Println("confirm handler: waiting on outstanding confirmations", len(confirms))

	CheckConfirmations(confirms)

	for k, v := range confirms {
		select {
		case <-v.Done():
			fmt.Println("confirm handler: confirmed delivery with tag: ", k)
			delete(confirms, k)
		case <-time.After(time.Second):
			fmt.Println("confirm handler: did not receive confirmation for tag：", k)
		}
	}

	outstandingConfirmationCount := len(confirms)
	if outstandingConfirmationCount > 0 {
		fmt.Println("confirm handler: exiting with outstanding confirmations", outstandingConfirmationCount)
	} else {
		fmt.Println("confirm handler: done waiting on outstanding confirmations")
	}
}
